package com.dark.kafka.trident;

/**
 * Created by tengxue on 16-7-6.
 */

import backtype.storm.generated.StormTopology;
import backtype.storm.spout.SchemeAsMultiScheme;
import storm.kafka.BrokerHosts;
import storm.kafka.StringScheme;
import storm.kafka.ZkHosts;
import storm.kafka.trident.OpaqueTridentKafkaSpout;
import storm.kafka.trident.TridentKafkaConfig;
import storm.trident.Stream;
import storm.trident.TridentTopology;

public class KafkaTridentTopologyBuilder {

    public static StormTopology buildTopology(String zkHostsStr, String topic){
        TridentTopology topology = new TridentTopology();
        BrokerHosts zk = new ZkHosts("localhost");
        TridentKafkaConfig spoutConf = new TridentKafkaConfig(zk, "log4j-test");
        spoutConf.scheme = new SchemeAsMultiScheme(new StringScheme());
        OpaqueTridentKafkaSpout spout = new OpaqueTridentKafkaSpout(spoutConf);


        Stream eventStream = topology.newStream("events", spout);

        return topology.build();
    }
}
